package com.sam.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author: wm.xue
 * @date: 2021/11/8 14:08
 * @description: 自定义拦截器
 */
public class _04_Producer_MyInterceptors {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.100.17.117:9092");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.ACKS_CONFIG, "1");
        // 设置重试次数
        configs.put(ProducerConfig.RETRIES_CONFIG, 3);
        // 每次连接请求最大请求数量
        configs.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
        // 设置自定义拦截器
        configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.sam.kafka.interceptor.MyInterceptorOne,com.sam.kafka.interceptor.MyInterceptorTwo,com.sam.kafka.interceptor.MyInterceptorThree");


        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);

        ProducerRecord<String,String> record = new ProducerRecord<String,String>("tp_04_demo",
                "1", "测试自定义拦截器");

        Future<RecordMetadata> future = producer.send(record);
        RecordMetadata metadata = future.get();
        System.err.println("发送消息成功：");
        System.out.println(metadata.toString());

    }

}
